package com.aabte.commons.redis.hash;

import com.aabte.commons.redis.RedisAdapter;
import com.google.common.hash.Funnel;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

/**
 * @author Daniel
 * @version 1.0
 * @date 2020/3/21
 */
@Data
@Slf4j
public class BloomFilter<T> {

  private RedisAdapter redisAdapter;
  private BloomFilterHelper<T> bloomFilterHelper;

  public static <T> BloomFilter<T> create(
      RedisAdapter redisAdapter, Funnel<T> funnel, int expectedInsertions, double fpp) {
      BloomFilterHelper<T> bloomFilterHelper = new BloomFilterHelper<T>(funnel, expectedInsertions, fpp);
      return new BloomFilter<T>(redisAdapter, bloomFilterHelper);
  }

  public BloomFilter(RedisAdapter redisAdapter, BloomFilterHelper<T> bloomFilterHelper) {
    this.redisAdapter = redisAdapter;
    this.bloomFilterHelper = bloomFilterHelper;
  }

  public void put(String key, T value) {
    int[] offsets = bloomFilterHelper.murmurHashOffset(value);
    for (int offset : offsets) {
      boolean rs = redisAdapter.setbit(key, offset, true);
      if (!rs) {
        log.warn("redis set bit failed. key: {}, offset: {}", key, offset);
      }
    }
  }

  public boolean mightContain(String key, T value) {
    int[] offsets = bloomFilterHelper.murmurHashOffset(value);
    for (int offset : offsets) {
      if (!redisAdapter.getbit(key, offset)) {
        return false;
      }
    }
    return true;
  }
}
